package com.heima.kafka.listener;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.*;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.handler.annotation.SendTo;

import java.time.Duration;
import java.util.Date;

@EnableBinding(KafkaStreamListener.CountNumProcessor.class)
public class KafkaStreamListener {

    @StreamListener("count-num-stream-topic")
    @SendTo("count-num-stream-consumer")
    public KStream<String, String> process(KStream<String, String> input) {

        // 做了数据格式的转换，把key从value中拆分出来，方便下一步做分组聚合
        return input.map(new KeyValueMapper<String, String, KeyValue<String, Integer>>() {
                    @Override
                    public KeyValue<String, Integer> apply(String key, String value) {
                        System.out.println(new Date() + "-收到原始数据:" + value);
                        String[] split = value.split(":");
                        return new KeyValue<>(split[0], Integer.valueOf(split[1]));
                    }
                    // 做分组聚合
                }).groupByKey(Grouped.with(Serdes.String(), Serdes.Integer()))
                // 设置时间窗口, 十秒为单位
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)).grace(Duration.ZERO))
                // 设置聚合算法
                .aggregate(new Initializer<Integer>() {
                    @Override
                    public Integer apply() {
                        System.out.println(new Date() + "--时间窗口初始化");
                        return 0;
                    }
                }, new Aggregator<String, Integer, Integer>() {
                    // 时间窗口内的聚合
                    @Override
                    public Integer apply(String key, Integer value, Integer aggregate) {
                        System.out.println(new Date() + "时间窗口内的聚合, 当前key = " + key
                                + "当前value = " + value + "当前aggregate = " + aggregate);
                        return value + aggregate;
                    }
                }, Materialized.with(Serdes.String(), Serdes.Integer()))
                // 只输出最终的计算结果, 不输出中间计算过程的数据
                .suppress(Suppressed.untilWindowCloses(Suppressed.BufferConfig.unbounded()))
                // 重新修改结果的数据类型, 因为默认的key为时间窗口的类型
                .toStream().map(new KeyValueMapper<Windowed<String>, Integer, KeyValue<String, String>>() {
                    @Override
                    public KeyValue<String, String> apply(Windowed<String> stringWindowed, Integer value) {
                        String key = stringWindowed.key();
                        System.out.println("时间窗口聚合结束, key = " + key + "value = " + value);

                        return new KeyValue<>(key, "" + value);
                    }
                });

    }

    public interface CountNumProcessor {
        @Input("count-num-stream-topic")
        KStream<String, String> input();

        @Output("count-num-stream-consumer")
        KStream<String, String> output();
    }
}
